package cn.doitedu.rpc

import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import com.typesafe.config.ConfigFactory

import scala.collection.mutable

import scala.concurrent.duration._

class Master extends Actor {


  //用于保存Worker信息的HashMap
  val idToWorkerInfo = new mutable.HashMap[String, WorkerInfo]()

  //在调用receive方法之前，一定会调用一次
  override def preStart(): Unit = {
    //启动定时器，定期检测超时的Worker，将超时的Worker异常
    //导入隐式转换
    import context.dispatcher
    context.system.scheduler.schedule(0 millisecond, Master.CHECK_HEARTBEAT_INTERVAL millisecond, self, CheckHeartbeat)
  }

  //receive用于接收其他Actor（也包括自己）发送的消息
  //Receive是偏函数类型，即PartialFunction[Any, Unit]
  override def receive: Receive = {

    //Worker向Master发送的注册消息
    case RegisterWorker(workerId, memory, cores) => {
      println(s"workerId: $workerId, memory: $memory, cores: $cores")
      //将Worker的信息封装起来然后保存起来
      val workerInfo = new WorkerInfo(workerId, memory, cores)
      //将Worker的信息保存起来(保证到可变的HashMap中)
      idToWorkerInfo(workerId) = workerInfo
      //向Worker返回注册成功的消息
      sender() ! RegisteredWorker
    }

    //Work发送个Master的消息消息
    case Heartbeat(workerId) => {
      //println(s"work : $workerId 发送了心跳消息！")
      //更新对应WorkerId的WorkerInfo的上一次心跳时间
      if (idToWorkerInfo.contains(workerId)) {
        //根据心跳发送过来的WorkerID，到Worker的注册信息中，找到对应的WorkerInfo
        val workerInfo = idToWorkerInfo(workerId)
        //更新最近心跳时间
        workerInfo.lastHeartbeatTime = System.currentTimeMillis()
      }
    }

    //Master自己发送给自己的定期执行心跳检测的消息
    case CheckHeartbeat => {

      //过滤出有超时的Worker
      val deadWorkers = idToWorkerInfo
        .values
        .filter(w => System.currentTimeMillis() - w.lastHeartbeatTime > 20000)
      //移除掉超时的Worker
//      for (deadWorker <- deadWorkers) {
//        //从idToWorkerInfo这个HashMap中异常对应的WorkerInfo
//        idToWorkerInfo -= deadWorker.workerId
//      }
      deadWorkers.foreach(w => {
        idToWorkerInfo -= w.workerId
      })

      println(s"current alive worker size: ${idToWorkerInfo.size}")

    }

  }
}

object Master {

  val MASTER_ACTOR_SYSTEM_NAME = "MasterActorSystem"
  val MASTER_ACTOR_NAME = "MasterActor"
  val CHECK_HEARTBEAT_INTERVAL = 20000

  def main(args: Array[String]): Unit = {

    val hostname = args(0)
    val port = args(1).toInt

    //参数的含义
    //akka.actor.provider 远程通信的实现方式
    //akka.remote.netty.tcp.hostname 绑定的地址
    //akka.remote.netty.tcp.port 绑定的端口
    val configStr =
    s"""
       |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
       |akka.remote.netty.tcp.hostname = $hostname
       |akka.remote.netty.tcp.port = $port
       |""".stripMargin
    //1.创建ActorSystem（用于创建并管理Actor的，一个进程中有一个即可，是单例的）
    //创建ActorSystem前要解析参数
    val config = ConfigFactory.parseString(configStr)
    //调用ActorSystem的apply方法，传入ActorSystem的名称和配置信息
    val actorSystem: ActorSystem = ActorSystem.apply(MASTER_ACTOR_SYSTEM_NAME, config)

    //2.创建Actor(使用反射创建的指定Actor类型的实例)
    //调用actorSystem的actorOf方法创建Actor，传入Actor的类型，Actor的名称
    actorSystem.actorOf(Props[Master], MASTER_ACTOR_NAME)

    //3.调用方法发消息（异步消息）
    //给指定的Actor发消息，！是一个方法
    //masterActor ! "hello"

  }


}
